package com.bigdata.sql

import org.apache.spark.sql._

import java.util.Properties

/**
 * 总结：
 * SparkSQL支持非常丰富的数据源，可以从文件，db中读取数据
 * 写出到文件、db中
 * 数据分析：
 * 现在有一批文件，我想要从文件中进行计算并提取结果
 * demo：把excel中的数据导出为csv，然后交由spark来计算
 * Spark可以用来分析本地类似excel的表格文件,一个表格只能存储1G的数据
 *
 * 假如我现在有50G的数据，并且每天的增量数据也有50G
 * 现在想用要Spark对这些数据做分析
 * 每天的增量数据存储在HDFS中，使用Spark分析HDFS中的文本数据
 *
 * 我们的大数据一定是存储在HDFS中的，使用spark计算HDFS中的数据，
 * 直接基于HDFS开发，很繁琐，想要简化开发
 * 我想要直接给Spark一个SQL语句，那么Spark就知道去这个sql查询的表所对应的HDFS文件中查询
 * 后续的所有操作只需要写SQL即可。
 * 我们需要引入一个组件：hive 的 metastore 服务！！！！
 */
object Demo4DataSource2 {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .appName("source")
      .master("local")
      .config("spark.sql.shuffle.partitions", 1)
      .getOrCreate()

    spark.sparkContext.setLogLevel("error")

    import org.apache.spark.sql.functions._
    import spark.implicits._

    /*
    大数据领域，历史数据一般会存储在HDFS，使用spark做数仓的离线分析
    数据源的来源就是HDFS，指定前缀hdfs://node01:9000即可
    这里就直接使用本地文件模拟
     */
    val studentDF: DataFrame = spark
      .read
      .format("csv")
      // csv文件只有数据，没有header，需要自己指定schema，指定列名和列的类型
      .schema("id STRING,name STRING,age INT,gender STRING, clazz STRING")
      .option("sep", ",")
      .load("hdfs://node01:9000/data/student")

    studentDF.printSchema()
    studentDF.show()

    /**
     * 将数据保持为csv格式
     * 按照班级分组，得到班级的人数,并将结果写入到指定目录
     */
    val stuCountDF: DataFrame = studentDF
      .groupBy($"clazz")
      .agg(count("clazz") as "count")

    // 把结果进行保存，可以直接存储在HDFS
    stuCountDF
      .write
      .format("csv")
      .option("sep", ",")
      .mode(SaveMode.Overwrite)
     .save("hdfs://node01:9000/data/csv_write")

  }
}
